package PassengerVehicle

import java.io.{BufferedReader, FileReader}
import java.util.Properties

import org.apache.spark.sql.{DataFrame, SparkSession}

object ProvinceInOut {

  def main(args: Array[String]): Unit = {
    //创建sparkSession
    val spark: SparkSession = SparkSession.builder().appName("Province").master("local[4]").getOrCreate()
    //创建properties对象 设置连接mysql的信息
    val prop: Properties = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "123456")

    val reader = new FileReader("D:/projects/spark/src/main/resource/result.txt")
    val bufferReader = new BufferedReader(reader)
    val dict = bufferReader.readLine()
    bufferReader.close()
    reader.close()

    //读取mysql数据
    val mysqlDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/localsql", "zhjg_jsy_ticket_smz_curmonth", prop)
    import spark.sql
    mysqlDF.createOrReplaceTempView("passenger")
    val s = sql("select XL,FCRQSJ,SCKCCPHM,DDZ,CC from passenger")
      .filter("SCKCCPHM!=''")

    // IN 省内  OUT 省外
    val result = s.rdd.map(item => {
      val xl = item.getString(0)
      val ddz = item.getString(3)
      if (matchDDZ(xl, ddz, dict)) {
        ("IN", 1)
      } else {
        ("OUT", 1)
      }
    }).reduceByKey((x, y) => x + y)

    result.foreach(x => {
      println(x._1 + "," + x._2)
    })

    spark.stop()
  }

  def matchDDZ(xl: String, ddz: String, dict: String): Boolean = {
    val reg1 = "[0-9]"
    val reg2 = "（[\\u4e00-\\u9fa5]*）"
    val reg3 = "[(][0-9\\u4e00-\\u9fa5]*[)]"
    val reg4 = "服务站|下道车|机场|国际机场|车站|汽车站|长途车站|长途汽车站|中心站|中心|高速学生|高速|补|路口"

    val xllen = xl.split("-").length
    if (dict.indexOf(ddz.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      return true
    } else if (xllen == 2 && dict.indexOf(xl.split("-")(1).replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      return true
    } else if (xllen == 1 && dict.indexOf(xl.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      return true
    } else {
      return false
    }
  }
}
